package regular;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;

public class EvalSecondMapper extends
		Mapper<Text, Text, IntWritable, NodeWritable> {

	public void map(Text key, Text value, Context context) throws IOException,
			InterruptedException {

		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);
		
		Initial.CONF = context.getConfiguration();

		ArrayList<Node> listNode = new ArrayList<Node>();
		LoadData(value, listNode, gq.us.stateName, gq.ut.stateName);
		
		NodeWritable nodes = new NodeWritable();
		nodes.listNode = listNode;

		context.write(new IntWritable(1), nodes);
	}

	public static void LoadData(Text value, ArrayList<Node> listNode,
			String startNode, String terminalNode) {
		try {

			ArrayList<Node> listInNode = Ultility
					.getListNode(Initial.PATH_INPUT_NODE_CACHE);
			ArrayList<Node> listOutNode = Ultility
					.getListNode(Initial.PATH_OUTPUT_NODE_CACHE);

			listInNode.add(new Node(Integer.parseInt(terminalNode)));
			listOutNode.add(new Node(Integer.parseInt(startNode)));
//			
//			for(Node n: listInNode)
//				System.out.println(n.id);
//			System.out.println("==========");
//			for(Node n: listOutNode)
//				System.out.println(n.id);

			byte[] bytes = value.getBytes();
			ByteArrayInputStream arr = new ByteArrayInputStream(bytes);
			BufferedReader bNode = new BufferedReader(
					new InputStreamReader(arr));
			String strLine = null;
			while ((strLine = bNode.readLine()) != null) {
				String[] node = strLine.split("\t");
				Node ni = new Node(Integer.parseInt(node[0]));

				if (listOutNode.contains(ni)) {

					for (int i = 1; i < node.length; i++) {
						if (node[i].equals("true"))
							node[i] = terminalNode;
						if (listInNode.contains(new Node(Integer
								.parseInt(node[i]))))
							ni.neighbors.add(Integer.parseInt(node[i]));
					}
					listNode.add(ni);
				}
			}
			bNode.close();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

class NodeWritable implements Writable {
	ArrayList<Node> listNode;

	public NodeWritable() {
		listNode = new ArrayList<Node>();
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		listNode.clear();
		int size = in.readInt();
		for (int i = 0; i < size; i++) {
			Node n = new Node();
			n.readFields(in);
			listNode.add(n);
		}
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(listNode.size());
		for (Node n : listNode) {
			n.write(out);
		}
	}
}

// Vertex define
class Node implements Writable {
	int id;
	boolean visit = false;
	ArrayList<Integer> neighbors;

	public Node() {
		neighbors = new ArrayList<Integer>();
	}

	public Node(int id) {
		this.id = id;
		this.neighbors = new ArrayList<Integer>();
	}

	public Node(int id, ArrayList<Integer> neighbors) {
		this.id = id;
		this.neighbors = neighbors;
	}

	@Override
	public boolean equals(Object obj) {
		boolean check = false;
		if (obj == null)
			return false;
		else if (obj == this)
			return true;
		else {
			if (obj instanceof Node && obj.getClass() == this.getClass()) {
				final Node t = (Node) obj;
				check = this.id == t.id;
			}
		}
		return check;
	}

	@Override
	public int hashCode() {
		int hash = 7;
		hash = (37 * hash) + id;
		return hash;
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		neighbors.clear();
		id = in.readInt();
		int size = in.readInt();
		for (int i = 0; i < size; i++) {
			neighbors.add(in.readInt());
		}
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(id);
		out.writeInt(neighbors.size());
		for (int n : neighbors)
			out.writeInt(n);
	}

}
